/**
 * Copyright 2020 Inc.
 **/
package com.myz.distributed.transaction.rocketmq.consumer;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.myz.distributed.transaction.rocketmq.common.TransactionService;
import com.myz.distributed.transaction.rocketmq.common.TransferRecordEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;

/**
 * 事务消息的监听器
 *
 * @author maoyz0621 on 2020/8/12
 * @version v1.0
 */
@RocketMQMessageListener(topic = "${rocketmq.topic.transaction.mysql}",
        consumerGroup = "${rocketmq.topic.transaction.mysql}",
        customizedTraceTopic = "${rocketmq.consumer.customized-trace-topic}")
@Component
@Slf4j
public class MySqlConsumerTransactionListener implements RocketMQListener<MessageExt> {

    @Resource
    TransactionService transactionService;
    @Resource
    ObjectMapper objectMapper;

    @Override
    public void onMessage(MessageExt message) {
        log.info("MySqlConsumerTransactionListener{}, {}, {}", message.getMsgId(), message.getTopic(), new String(message.getBody()));
        if (message.getReconsumeTimes() > 2) {
            log.error("超过重试次数");
            return;
        }

        try {
            TransferRecordEntity transferRecord = objectMapper.readValue(message.getBody(), TransferRecordEntity.class);
            transactionService.increaseMoney(transferRecord.getToUserId(), transferRecord.getChangeMoney());
        } catch (IOException e) {
            throw new RuntimeException();
        }
    }
}